package com.spark.ml.demo

import org.apache.hadoop.fs.FileSystem
import org.apache.log4j.{Level, Logger}
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.sql.{Dataset, Row, SparkSession, expressions, functions}
import org.apache.spark.{SparkConf, SparkContext}
import org.jblas.DoubleMatrix

import scala.collection.mutable

object ALSDemo {
  // 定义相似度函数
  def cosineSimilarity(vec1: DoubleMatrix, vec2: DoubleMatrix): Double = {
    vec1.dot(vec2) / (vec1.norm2() * vec2.norm2())
  }

  def loadModel(sc: SparkContext, path: String): MatrixFactorizationModel = {
    MatrixFactorizationModel.load(sc, path)
  }

  def getDataSet(ss: SparkSession, file: String): Dataset[Row] = {
    //val hadoopConf = sc.hadoopConfiguration
    //val fs = FileSystem.get(hadoopConf)
    ss.read.json(file)
  }

  case class UidAndADid(val userId: Int, val adId: Int)

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("als").setMaster("local")
    val sc = new SparkContext(conf)
    Logger.getRootLogger.setLevel(Level.WARN)
    println("read history data ")
    val ss = SparkSession.builder().config(conf).getOrCreate()
    val fileName = "/Users/qifei/work/logs.1513069889111"
    val input = getDataSet(ss, fileName)
    val distinctInput = input.filter(functions.col("counterid").isin(
      "ad_adview",
      "ad_adcomplete",
      "ad_adclick")
    ).select("userid", "appid", "rowkey",
      "counterid"
    ).filter(functions.col("userid").notEqual("--")).distinct()
    distinctInput.persist()

    val result = distinctInput.select(
      "userid", "appid", "rowkey",
      "counterid"
    ).withColumn("counterid",
      functions.when(functions.col("counterid").isin("ad_adview"), 1d).otherwise(
        functions.when(functions.col("counterid").isin("ad_adcomplete"), 2d).otherwise(
          functions.when(functions.col("counterid").isin("ad_adclick"), 4d)
        )
      ))

    val sum = result.groupBy("userid", "appid", "rowkey").sum("counterid").
      withColumnRenamed("sum(counterid)", "rate").sort(functions.desc("rate"))
    println("原始数据中的20条数据")
    sum.show(20)
    var intUserId = 1
    var intAdId = 1
    val userId = new mutable.HashMap[String, Int]()
    val adId = new mutable.HashMap[String, Int]()
    //trans to int

    val ratings = sum.rdd.map(r => {

      if (!userId.contains(r.getAs(0))) {
        intUserId += 1
        userId.put(r.getAs(0), intUserId)
      }

      if (!adId.contains(r.getAs(2))) {
        intAdId += 1
        adId.put(r.getAs(2), intAdId)
      }
      Rating(userId.get(r.getAs(0)).get, adId.get(r.getAs(2)).get, r.getAs(3))
    })
    println("转换后的20条数据")
    ratings.take(20).foreach(println)
    //ratings.foreach(println)
    //create model
    val rank = 10
    val numIterations = 10
    val model = ALS.train(ratings, rank, numIterations, 0.01)

    //predict result
    val usersProducts = ratings.map {
      case Rating(user, product, rate) =>
        (user, product)
    }
    val predictions = model.predict(usersProducts).map {
      case Rating(user, product, rate) =>
        ((user, product), rate)
    }
    val ratesAndPreds = ratings.map {
      case Rating(user, product, rate) =>
        ((user, product), rate)
    } join (predictions)

    val MSE = ratesAndPreds.map {
      case ((user, product), (r1, r2)) =>
        val err = (r1 - r2)
        err * err
    }.mean()
    //recommend product for old user，获取某个用户推荐topN 广告，以及其评分
    val p = model.recommendProducts(15, 3)
    println("给用户(id=15)推荐三个它评分最高的广告")
    println(p.mkString("\n"))
    //获取某个用户所有打分记录,比如用户1
    val productForUser = ratings.keyBy(_.user).lookup(1)
    println(productForUser.mkString("\n"))
    //推荐前，需要过滤掉已观看过的广告

    //广告推荐
    // 选定id为2的广告
    val itemId = 2
    // 获取该物品的隐因子向量
    val itemFactor = model.productFeatures.lookup(itemId).head
    // 将该向量转换为jblas矩阵类型
    val itemVector = new DoubleMatrix(itemFactor)
    // 计算广告2 与其他广告的相似度
    val sims = model.productFeatures.map { case (id, factor) =>
      val factorVector = new DoubleMatrix(factor)
      val sim = cosineSimilarity(factorVector, itemVector)
      (id, sim)
    }
    // 获取与广告2最相似的3个广告
    val sortedSims = sims.top(3)(Ordering.by[(Int, Double), Double] { case (id, similarity) => similarity })
    // 打印结果
    //sortedSims.slice(1, 11).map { case (id, sim) => (id, sim) }.mkString("\n")

    // 获取和广告2最相似的3个广告(含自己)
    val sortedSims2 = sims.top(3 + 1)(Ordering.by[(Int, Double), Double] { case (id, similarity) => similarity })
    // 再打印和2最相似的3个广告

    sortedSims2.slice(1, 11).map { case (id, sim) => (id, sim) }.mkString("\n")
    println("Mean Squared Error = " + MSE)

    //save model
    val modelPath = "/Users/qifei/work/machine"
    //model.save(sc,modelPath)
  }
}
